package spring.cloud.tasks.client;

import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.context.ApplicationContext;
import spring.cloud.tasks.Worker;
import spring.cloud.tasks.client.batch_task_processor.*;
import spring.cloud.tasks.client.util.Addresss;
import spring.cloud.tasks.common.utils.ApplicationContextUtils;
import spring.cloud.tasks.common.zookeeper.ZooKeeperBatchTaskDataManager;
import spring.cloud.tasks.common.zookeeper.ZooKeeperManager;

import java.io.FileNotFoundException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class BatchTaskWorker extends Worker {


    @Setter
    @Getter
    private String workerId;
    @Setter
    @Getter
    private String ip;
    @Setter
    @Getter
    private String hostName;
    @Setter
    @Getter
    protected ZooKeeperManager zooKeeperManager;
    @Setter
    @Getter
    private ZooKeeperBatchTaskDataManager zooKeeperBatchTaskDataManager;

    @Setter
    @Getter
    private ZooKeepersConnectAndWorkerTasksInitThread zooKeepersConnectAndWorkerTasksInitThread;
    @Setter
    @Getter
    volatile String initThreadErrorMessage = "";
    @Setter
    @Getter
    private Timer timer;
    @Setter
    @Getter
    private WorkerTimerTask workerTimerTask;
    @Setter
    @Getter
    private int timerInterval = 2000;
    @Setter
    @Getter
    volatile String pongpongErrorMessage = "";

    @Setter
    @Getter
    public boolean init = false;//zk初始化成功
    @Setter
    @Getter
    public boolean executeTasks = true;
    //
    protected Lock lock = new ReentrantLock();

    @Setter
    @Getter
    private List<String> taskIdList = new ArrayList<String>();
    //
    @Setter
    @Getter
    private Map<String, BatchTaskInfo> taskIdToBatchTaskInfoMap = new ConcurrentHashMap<String, BatchTaskInfo>();


    public BatchTaskWorker() {
        this.ip = Addresss.getLocalHostAddress();
        this.hostName = Addresss.getLocalHostName();
        ApplicationContext applicationContext = ApplicationContextUtils.getApplicationContext();
        Map<String, BatchTask> stringTaskDataProcessorMap = applicationContext.getBeansOfType(BatchTask.class);
        for (BatchTask taskDataProcessor : stringTaskDataProcessorMap.values()) {
            taskIdList.add(taskDataProcessor.getTaskId());
        }
    }

    public ZooKeeperBatchTaskDataManager getZooKeeperBatchTaskDataManager() {
        if (this.zooKeeperBatchTaskDataManager == null) {
            throw new NullPointerException("zooKeeperBatchTaskDataManager");
        }
        return zooKeeperBatchTaskDataManager;
    }


    public void init() throws Exception {
        Properties properties = new Properties();
        this.init(properties);
    }

    /**
     * 初始化
     */
    public void init(Properties properties) throws Exception {

        //
        this.lock.lock();
        try {
            if (this.zooKeepersConnectAndWorkerTasksInitThread != null) {
                this.zooKeepersConnectAndWorkerTasksInitThread.stopInitThread();
                this.zooKeepersConnectAndWorkerTasksInitThread = null;
            }
            if (timer != null) {
                timer.cancel();
                timer = null;
            }

            this.zooKeeperBatchTaskDataManager = null;
            if (this.zooKeeperManager != null) {
                this.zooKeeperManager.close();
            }
            //
            Client.setBatchTaskWorker(this);
            this.zooKeeperManager = new ZooKeeperManager(properties);
            zooKeepersConnectAndWorkerTasksInitThread = new ZooKeepersConnectAndWorkerTasksInitThread(this);
            zooKeepersConnectAndWorkerTasksInitThread.setName("ItemTaskWorker-zooKeepersConnectAndWorkerTasksInitThread");
            zooKeepersConnectAndWorkerTasksInitThread.start();
        } finally {
            this.lock.unlock();
        }
    }

    /**
     * zk连接成功后进行初始化
     */
    public void initTasksAfterZooKeeperConnected() throws Exception {
        this.zooKeeperManager.doInit();
        this.zooKeeperBatchTaskDataManager = new ZooKeeperBatchTaskDataManager(this.zooKeeperManager);
        List<String> taskIdList = this.zooKeeperBatchTaskDataManager.registerWorkerToPermittedTaskList(this);
        log.info("注册成功taskIdList:" + JSON.toJSONString(taskIdList));
        //下面进行worker和zk交互逻辑
        if (this.timer != null) {
            this.timer.cancel();
            this.timer = null;
        }
        this.timer = new Timer("ItemTaskWorker-Timer");
        //
        if (this.workerTimerTask != null) {
            this.workerTimerTask = null;
        }
        this.workerTimerTask = new WorkerTimerTask(this);
        this.timer.schedule(this.workerTimerTask, 2000, this.timerInterval);
    }


    /**
     * zk连接不上才重启
     */
    public void reStart() throws Exception {
        try {
            if (this.timer != null) {
                if (this.workerTimerTask != null) {
                    this.workerTimerTask.cancel();
                    this.workerTimerTask = null;//对象并行处理模型
                }
                this.timer.purge();
            }
            //

            stopBatcherList();
            //
            if (this.zooKeeperManager != null) {
                this.zooKeeperManager.close();
            }
            this.workerId = null;
            //
            this.init();
        } catch (Throwable e) {
            log.error("" + e.getMessage(), e);
        }
    }

    public void stop() throws Exception {
        try {
            lock.lock();
            this.executeTasks = false;
            if (this.zooKeepersConnectAndWorkerTasksInitThread != null) {
                this.zooKeepersConnectAndWorkerTasksInitThread.stopInitThread();
            }
            if (this.timer != null) {
                if (this.workerTimerTask != null) {
                    this.workerTimerTask.cancel();
                    this.workerTimerTask = null;
                }
                this.timer.cancel();
                this.timer = null;
            }

            stopBatcherList();
            if (this.zooKeeperManager != null) {
                this.zooKeeperManager.close();
            }
            if (this.getZooKeeperBatchTaskDataManager() != null) {
                try {
                    ZooKeeper zooKeeper = this.getZooKeeperBatchTaskDataManager().getZooKeeper();
                    if (zooKeeper != null) {
                        zooKeeper.close();
                    }
                } catch (Exception e) {
                    log.error("stopAll zookeeper getZooKeeper", e);
                }
            }
            this.workerId = null;
            log.info("stopAll");
        } catch (Throwable e) {
            log.error("stopAll" + e.getMessage(), e);
        } finally {
            lock.unlock();
        }
    }

    public void pongpong() throws Exception {
        this.lock.lock();
        try {
            WorkerInfo workerInfo = null;
            boolean fileNotFoundException = false;
            try {
                workerInfo = this.getZooKeeperBatchTaskDataManager().loadWorkerInfo(this.getWorkerId());//总体信息
            } catch (FileNotFoundException e) {
                log.error("获取workerInfo失败", e);
                fileNotFoundException = true;
            }
            if (fileNotFoundException == true) {//被清理了
                try {
                    stopBatcherList();
                    this.getZooKeeperBatchTaskDataManager().unRegisterWorker(this);//卸载调度器
                } finally {
                    rebuildTaskProcessorListWhenAllowExecuteTasks();
                }
            } else {
                if (workerInfo.isExecuteTasks() == false) {

                    stopBatcherList();
                } else {
                    //正常
                    rebuildBatcherList();
                }
            }
        } finally {
            this.lock.unlock();
        }
    }

    /**
     * 完全重新构建
     */
    public void stopBatcherList() throws Exception {
        for (String batchTaskId : taskIdToBatchTaskInfoMap.keySet()) {
            BatchTaskInfo batchTaskInfo = taskIdToBatchTaskInfoMap.remove(batchTaskId);
            if (batchTaskInfo != null) {
                batchTaskInfo.stop();
            }
            taskIdToBatchTaskInfoMap.remove(batchTaskId);
        }
    }

    public void rebuildTaskProcessorListWhenAllowExecuteTasks() throws Exception {
        //只支持worker允许运行的情况下才进行调用
        List<String> taskIdList = this.getZooKeeperBatchTaskDataManager().registerWorkerToPermittedTaskListWhenPongPong(this);
        //这个是允许存活的任务,排开这些之外的需要进行清理
        this.stopTaskProcessorListNotTaskIdList(taskIdList);
        this.rebuildTaskProcessorListByWorkerId(taskIdList);
    }

    private void stopTaskProcessorListNotTaskIdList(List<String> taskIdList) {
        for (String batchTaskId : taskIdToBatchTaskInfoMap.keySet()) {
            if (!taskIdToBatchTaskInfoMap.containsKey(taskIdList)) {
                BatchTaskInfo batchTaskInfo = taskIdToBatchTaskInfoMap.remove(batchTaskId);
                if (batchTaskInfo != null) {
                    batchTaskInfo.stop();
                }
                taskIdToBatchTaskInfoMap.remove(batchTaskId);
            }
        }
    }

    private void rebuildTaskProcessorListByWorkerId(List<String> taskIdList) throws Exception {
        for (String taskId : taskIdList) {
            if (!taskIdToBatchTaskInfoMap.containsKey(taskId)) {
                BatchTaskInfo batchTaskInfo = new BatchTaskInfo();
                batchTaskInfo.setTaskId(taskId);
                taskIdToBatchTaskInfoMap.put(taskId, batchTaskInfo);
            }
        }
        Date now = new Date();
        long nowLong = now.getTime();
        for (String taskId : taskIdList) {
            //补齐创建对应的处理线程
            BatchTaskInfo batchTaskInfo = taskIdToBatchTaskInfoMap.get(taskId);
            List<String> taskInstanceIdList = zooKeeperBatchTaskDataManager.getTaskInstanceIdList(taskId, workerId);
            //清理多余的实例
            batchTaskInfo.stop(taskInstanceIdList);
            //
            for (String taskInstanceId : taskInstanceIdList) {
                BatchTaskWorkerInstanceContext batchTaskWorkerInstanceContext = batchTaskInfo.taskInstanceIdToWorkerTaskBatchInfoMap.get(taskInstanceId);
                if (batchTaskWorkerInstanceContext == null) {
                    batchTaskWorkerInstanceContext = new BatchTaskWorkerInstanceContext();
                    batchTaskWorkerInstanceContext.setTaskId(taskId);
                    batchTaskWorkerInstanceContext.setTaskInstanceId(taskInstanceId);
                    batchTaskInfo.taskInstanceIdToWorkerTaskBatchInfoMap.put(taskInstanceId, batchTaskWorkerInstanceContext);
                }

                final BatchTaskWorkerInstanceServerInfo batchTaskWorkerInstanceServerInfo = zooKeeperBatchTaskDataManager.getWorkerTaskInstanceBatchInfo(taskId, taskInstanceId);
                batchTaskWorkerInstanceContext.setBatchTaskWorkerInstanceServerInfo(batchTaskWorkerInstanceServerInfo); //设置最新的任务实例信息
                if (batchTaskWorkerInstanceServerInfo == null) {
                    batchTaskInfo.stop(taskInstanceId);//有可能临时被删除了

                } else {
                    BatchTaskWorkerInstanceClientInfo batchTaskWorkerInstanceClientInfo = batchTaskWorkerInstanceContext.getBatchTaskWorkerInstanceClientInfo();
                    if (batchTaskWorkerInstanceClientInfo == null) {
                        batchTaskWorkerInstanceClientInfo = new BatchTaskWorkerInstanceClientInfo();
                        batchTaskWorkerInstanceClientInfo.setTaskId(taskId);
                        batchTaskWorkerInstanceClientInfo.setTaskInstanceId(taskInstanceId);
                        batchTaskWorkerInstanceContext.setBatchTaskWorkerInstanceClientInfo(batchTaskWorkerInstanceClientInfo);
                    }
                    //
                    String batchCreaterRun = batchTaskWorkerInstanceServerInfo.getBatchCreaterRun();
                    if ("Y".equals(batchCreaterRun) && batchTaskWorkerInstanceServerInfo.needRunOfBatchCreater() && workerId.equals(batchTaskWorkerInstanceServerInfo.getBatchCreaterRunWorkerId())) {
                        BatchTaskWorkerInstanceClientInfo.TaskThreadInfo taskThreadInfo = batchTaskWorkerInstanceClientInfo.getTaskCreaterThreadInfo();
                        boolean needCreateTaskDealThread = false;
                        if (taskThreadInfo == null) {
                            needCreateTaskDealThread = true;
                        } else {
                            if (now.getTime() - taskThreadInfo.getLastUpdateDate().getTime() > 1000 * 60) {
                                needCreateTaskDealThread = true;
                            }
                        }
                        if (needCreateTaskDealThread) {
                            //设置心跳信息
                            String threadId = BatchCreater.class.getName() + "-" + taskId + "-" + taskInstanceId + "-" + nowLong + "-" + UUID.randomUUID().toString();
                            BatchTaskInstanceExecuterThreadInfo batchTaskInstanceExecuterThreadInfo = new BatchTaskInstanceExecuterThreadInfo();
                            batchTaskInstanceExecuterThreadInfo.setTaskId(taskId);
                            batchTaskInstanceExecuterThreadInfo.setTaskInstanceId(taskInstanceId);
                            batchTaskInstanceExecuterThreadInfo.setThreadId(threadId);
                            batchTaskInstanceExecuterThreadInfo.setThreadType(BatchCreater.class.getName());
                            BatchTaskInstanceExecuterThread thread = new BatchTaskInstanceExecuterThread() {
                                @Override
                                public void run() {
                                    BatchCreater batchCreater = null;
                                    setBatchTaskExecuter(batchCreater);
                                    batchCreater.batchCreate(getBatchTaskWorker(), getTaskInstanceThreadInfo(), getParameter());
                                }
                            };
                            thread.setBatchTaskWorker(this);
                            thread.setTaskInstanceThreadInfo(batchTaskInstanceExecuterThreadInfo);
                            thread.setParameter(batchTaskWorkerInstanceServerInfo.getBatchResultCollecterRunParameter());
                            thread.setDaemon(true);
                            thread.setName(taskId);
                            thread.start();
                            //
                            taskThreadInfo = new BatchTaskWorkerInstanceClientInfo.TaskThreadInfo();
                            taskThreadInfo.setTaskId(taskId);
                            taskThreadInfo.setTaskInstanceId(taskInstanceId);
                            taskThreadInfo.setThreadId(threadId);
                            taskThreadInfo.setLastUpdateDate(now);
                            batchTaskWorkerInstanceClientInfo.setTaskCreaterThreadInfo(taskThreadInfo);
                        }

                    } else {
                        //如之前存在则需要注销
                        batchTaskWorkerInstanceClientInfo.setTaskCreaterThreadInfo(null);
                    }
                    String batchExecuterRun = batchTaskWorkerInstanceServerInfo.getBatchExecuterRun();
                    int threadExecuteCountPerBatcher = batchTaskWorkerInstanceServerInfo.getThreadExecuteCountPerBatcher();
                    if ("Y".equals(batchExecuterRun) && batchTaskWorkerInstanceServerInfo.needRunOfBatchExecuter() && batchTaskWorkerInstanceServerInfo.getBatchCreaterRunWorkerIdList().contains(workerId)) {
                        List<BatchTaskWorkerInstanceClientInfo.TaskThreadInfo> taskThreadInfoList = batchTaskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList();
                        List<BatchTaskWorkerInstanceClientInfo.TaskThreadInfo> taskThreadInfoList1 = new ArrayList<BatchTaskWorkerInstanceClientInfo.TaskThreadInfo>();
                        for (BatchTaskWorkerInstanceClientInfo.TaskThreadInfo taskThreadInfo : taskThreadInfoList) {
                            if (!(now.getTime() - taskThreadInfo.getLastUpdateDate().getTime() > 1000 * 60)) {
                                taskThreadInfoList1.add(taskThreadInfo);
                            }
                        }
                        taskThreadInfoList.clear();
                        batchTaskWorkerInstanceClientInfo.setTaskExecuterThreadInfoList(taskThreadInfoList1);
                        int size = taskThreadInfoList1.size();
                        int taskThreadSize = threadExecuteCountPerBatcher - size;
                        if (taskThreadSize > 0) {
                            //线程弥补
                            for (int i = 1; i <= taskThreadSize; i++) {
                                //设置心跳信息
                                String threadId = BatchExecuter.class.getName() + "-" + taskId + "-" + taskInstanceId + "-" + nowLong + "-" + UUID.randomUUID().toString();
                                BatchTaskInstanceExecuterThreadInfo batchTaskInstanceExecuterThreadInfo = new BatchTaskInstanceExecuterThreadInfo();
                                batchTaskInstanceExecuterThreadInfo.setTaskId(taskId);
                                batchTaskInstanceExecuterThreadInfo.setTaskInstanceId(taskInstanceId);
                                batchTaskInstanceExecuterThreadInfo.setThreadId(threadId);
                                batchTaskInstanceExecuterThreadInfo.setThreadType(BatchExecuter.class.getName());
                                BatchTaskInstanceExecuterThread thread = new BatchTaskInstanceExecuterThread() {
                                    @Override
                                    public void run() {
                                        BatchExecuter batchExecuter = null;
                                        setBatchTaskExecuter(batchExecuter);
                                        batchExecuter.executeBatch(getBatchTaskWorker(), getTaskInstanceThreadInfo(), getParameter());
                                    }
                                };
                                thread.setBatchTaskWorker(this);
                                thread.setTaskInstanceThreadInfo(batchTaskInstanceExecuterThreadInfo);
                                thread.setParameter(batchTaskWorkerInstanceServerInfo.getBatchResultCollecterRunParameter());
                                thread.setDaemon(true);
                                thread.setName(taskId);
                                thread.start();
                                //
                                BatchTaskWorkerInstanceClientInfo.TaskThreadInfo taskThreadInfo = new BatchTaskWorkerInstanceClientInfo.TaskThreadInfo();
                                taskThreadInfo.setTaskId(taskId);
                                taskThreadInfo.setTaskInstanceId(taskInstanceId);
                                taskThreadInfo.setThreadId(threadId);
                                taskThreadInfo.setLastUpdateDate(now);
                                batchTaskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().add(taskThreadInfo);
                            }
                        } else if (taskThreadSize < 0) {
                            if (threadExecuteCountPerBatcher > 0) {
                                while (batchTaskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().size() > threadExecuteCountPerBatcher) {
                                    batchTaskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().remove(batchTaskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().size() - 1);
                                }
                            }

                        }

                    } else {
                        batchTaskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().clear();

                    }
                    String batchResultCollecterRun = batchTaskWorkerInstanceServerInfo.getBatchCreaterRun();
                    if ("Y".equals(batchResultCollecterRun) && batchTaskWorkerInstanceServerInfo.needRunOfBatchResultCollecter() && workerId.equals(batchTaskWorkerInstanceServerInfo.getBatchResultCollecterRunWorkerId())) {
                        BatchTaskWorkerInstanceClientInfo.TaskThreadInfo taskThreadInfo = batchTaskWorkerInstanceClientInfo.getTaskResultCollecterThreadInfo();
                        boolean needCreateTaskDealThread = false;
                        if (taskThreadInfo == null) {
                            needCreateTaskDealThread = true;
                        } else {
                            if (now.getTime() - taskThreadInfo.getLastUpdateDate().getTime() > 1000 * 60) {
                                needCreateTaskDealThread = true;
                            }
                        }
                        if (needCreateTaskDealThread) {
                            //设置心跳信息
                            String threadId = BatchResultCollecter.class.getName() + "-" + taskId + "-" + taskInstanceId + "-" + nowLong + "-" + UUID.randomUUID().toString();
                            BatchTaskInstanceExecuterThreadInfo batchTaskInstanceExecuterThreadInfo = new BatchTaskInstanceExecuterThreadInfo();
                            batchTaskInstanceExecuterThreadInfo.setTaskId(taskId);
                            batchTaskInstanceExecuterThreadInfo.setTaskInstanceId(taskInstanceId);
                            batchTaskInstanceExecuterThreadInfo.setThreadId(threadId);
                            batchTaskInstanceExecuterThreadInfo.setThreadType(BatchResultCollecter.class.getName());
                            //
                            BatchTaskInstanceExecuterThread thread = new BatchTaskInstanceExecuterThread() {
                                @Override
                                public void run() {
                                    BatchResultCollecter batchResultCollecter = null;
                                    setBatchTaskExecuter(batchResultCollecter);
                                    batchResultCollecter.batchResultCollecte(getBatchTaskWorker(), getTaskInstanceThreadInfo(), getParameter());
                                }
                            };
                            thread.setBatchTaskWorker(this);
                            thread.setTaskInstanceThreadInfo(batchTaskInstanceExecuterThreadInfo);
                            thread.setParameter(batchTaskWorkerInstanceServerInfo.getBatchResultCollecterRunParameter());
                            thread.setDaemon(true);
                            thread.setName(taskId);
                            thread.start();
                            //
                            taskThreadInfo = new BatchTaskWorkerInstanceClientInfo.TaskThreadInfo();
                            taskThreadInfo.setTaskId(taskId);
                            taskThreadInfo.setTaskInstanceId(taskInstanceId);
                            taskThreadInfo.setThreadId(threadId);
                            taskThreadInfo.setLastUpdateDate(now);
                            //
                            batchTaskWorkerInstanceClientInfo.setTaskResultCollecterThreadInfo(taskThreadInfo);
                        }

                    } else {
                        batchTaskWorkerInstanceClientInfo.setTaskResultCollecterThreadInfo(null);
                    }
                }
            }


        }
    }

    private void rebuildBatcherList() {
    }

}

